package com.qf.smsplatform.gateway.thread;

import com.qf.smsplatform.common.constants.CacheConstants;
import com.qf.smsplatform.common.constants.ReportStateConstants;
import com.qf.smsplatform.common.model.StandardReport;
import com.qf.smsplatform.common.model.StandardSubmit;
import com.qf.smsplatform.gateway.client.CacheClient;
import com.qf.smsplatform.gateway.mq.DelayMessagePostProcessor;
import com.qf.smsplatform.gateway.mq.StatQueue;
import com.qf.smsplatform.gateway.netty4.entity.CmppDeliver;
import com.qf.smsplatform.gateway.netty4.entity.ReportMatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;

import static com.qf.smsplatform.gateway.config.RabbitConfig.QUEUE_DELAY_PER_MESSAGE_TTL_MSG_SMS_SEND;

/**
 * 多线程类
 * @author zjw
 */
public class SendReportThread implements Runnable {

    private final static Logger log = LoggerFactory.getLogger(SendReportThread.class);

    private AmqpTemplate rabbitTemplate;
    private CacheClient cacheClient;
    private String pushTopic;
    private String updateTopic;

    private StatQueue statqueue = StatQueue.getInstance();

    public SendReportThread(String pushTopic, String updateTopic, AmqpTemplate rabbitTemplate) {
        this.pushTopic = pushTopic;
        this.updateTopic = updateTopic;
        this.rabbitTemplate = rabbitTemplate;
    }
    public SendReportThread(String pushTopic, String updateTopic, AmqpTemplate rabbitTemplate, CacheClient cacheClient) {
        this.pushTopic = pushTopic;
        this.updateTopic = updateTopic;
        this.rabbitTemplate = rabbitTemplate;
        this.cacheClient = cacheClient;
    }

    @Override
    public void run() {
        while (true) {
            try {
                if (statqueue.size() > 0) {
                    // 获取第二次回调返回的对象msg
                    CmppDeliver deliver = (CmppDeliver) statqueue.remove();


                    //  获取运营商第二次回调放回的state，通过这个状态判断短信发送情况，返回DELIVRD代表发送成功
                    String stat = deliver.getStat();


                    // submit中的msgid，可以通过ReportMatch取出第一次回调时存储的submit对象
                    long msgid = deliver.getMsg_Id_DELIVRD();
                    ReportMatch reportMatch = ReportMatch.getInstance();
                    StandardSubmit submit = (StandardSubmit) reportMatch.remove(msgid);



                    StandardReport report = new StandardReport();
                    if("DELIVRD".equals(stat)){
                        // 短信发送成功
                        report.setState(ReportStateConstants.SUCCESS);
                        // TODO  限流…………
                        // TODO  扣费………… 扣数据库
                    }else{
                        // 短信发送失败
                        String errorMessage = deliver.getMsg_Content();
                        report.setState(ReportStateConstants.FAIL);
                        report.setErrorCode(errorMessage);
                        // TODO  扣费………… Redis的hincr
                    }

                    // 封装report
                    report.setSrcID(submit.getSrcSequenceId());
                    report.setMobile(submit.getDestMobile());
                    report.setMsgId(String.valueOf(msgid));
                    report.setClientID(submit.getClientID());
                    report.setSequenceId(submit.getSequenceId());

                    // 发送消息到状态报告TOPIC
                    Integer isReturnStatus = Integer.valueOf(cacheClient.hget(CacheConstants.CACHE_PREFIX_CLIENT + submit.getClientID(),"isreturnstatus"));
                    if(isReturnStatus == 1) {
                        rabbitTemplate.convertAndSend(pushTopic, report);
                    }
                    // 发送消息到队列并且设置消息的生存时间为10s
                    // 只要10s已过，消息变为死信，通过死信交换机路由到死信队列，死信队列就是 report_update_topic
                    rabbitTemplate.convertAndSend(QUEUE_DELAY_PER_MESSAGE_TTL_MSG_SMS_SEND, report, new DelayMessagePostProcessor(10 * 1000));
                } else {
                    sleep(2000);
                }
            } catch (Exception ex) {
                log.error(ex.getMessage(), ex);
                sleep(2000);
            }
        }
    }

    private void sleep(long time) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
    }
}
